package com.hadoop.mapreduce.mobilestatistics.reduce;

import com.hadoop.mapreduce.mobilestatistics.bean.MobileStatisticalData;
import com.hadoop.mapreduce.mobilestatistics.bean.MobileTrafficInformation;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;


/**
 * @author chenxiguo
 */
public class MobileReduce extends Reducer<Text, MobileTrafficInformation, MobileStatisticalData, Text> {


    @Override
    protected void reduce(Text key, Iterable<MobileTrafficInformation> values, Reducer<Text, MobileTrafficInformation,
            MobileStatisticalData, Text>.Context context) throws IOException, InterruptedException {
        Long downLinkTraffic = 0L;
        Long upLinkTraffic = 0L;
        for (MobileTrafficInformation original : values) {
            downLinkTraffic = downLinkTraffic + original.getDownLinkAmpacity();
            upLinkTraffic = upLinkTraffic + original.getUpstreamDownstreamTraffic();
        }
        Long totalFlow = downLinkTraffic + upLinkTraffic;
        MobileStatisticalData data = new MobileStatisticalData();
        data.setMobile(key.toString());
        data.setDownLinkTraffic(downLinkTraffic);
        data.setUpLinkTraffic(upLinkTraffic);
        data.setTotalFlow(totalFlow);
        context.write(data, null);
    }
}
